package com.pudaocredit.function;

import com.pudaocredit.Exception.NoTableException;
import com.pudaocredit.constans.HTableConst;
import com.pudaocredit.serializer.QueryHbaseCaihongTblUdfSerializer;
import com.pudaocredit.utils.HbaseConUtil;
import org.apache.hadoop.hbase.ClusterStatus;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hive.com.esotericsoftware.kryo.DefaultSerializer;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class QueryHbaseCaihongTblUdf2 extends UDF {
    private static org.slf4j.Logger logger = LoggerFactory.getLogger(QueryHbaseCaihongTblUdf2.class);
    Connection connection = null;

    String colfamily;

    public ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {
        /*try{
            connection = HbaseConUtil.getConnection("11.243.32.178,11.243.32.179,11.243.32.177", "2181", "/export/kerberos/1/hive.keytab", "/etc/krb5.conf", "hive/ocean@PUDAO.COM");
            System.out.println(connection);
            admin = connection.getAdmin();
            table = connection.getTable(TableName.valueOf(HTableConst.MOBILE_SHA256_TABLE));
            if (admin.tableExists(TableName.valueOf(HTableConst.MOBILE_SHA256_TABLE))){
                System.out.println(HTableConst.MOBILE_SHA256_TABLE+"存在");

            }
//        connection = HbaseConUtil.getConnection("11.243.32.178,11.243.32.179,11.243.32.177", "2181", KEYTAB_FILE, KERB_CONF, "hive/ocean@PUDAO.COM");
        }catch (Exception e){
            System.out.println("创建hbase连接失败");
            logger.info("创建hbase连接失败");
            e.printStackTrace();
        }*/


        if (objectInspectors.length != 3) {
            throw new UDFArgumentException("Input args length  erro!!!");
        } else if (!objectInspectors[0].getCategory().equals(ObjectInspector.Category.PRIMITIVE)) {
            throw new UDFArgumentTypeException(0, "Input args  type error!!");
        } else {
            return PrimitiveObjectInspectorFactory.javaStringObjectInspector;
        }
    }

    public String  evaluate(String str,String info_type,String hash_type) throws HiveException {
        System.out.println("参数str："+str+",信息类型为："+info_type+"，hash类型为"+hash_type);
        Table table = null;

        if (table == null){
            try{
                System.out.println("初始化hbase客户端.......");
                logger.info("初始化hbase客户端.......");
                connection = HbaseConUtil.getConnection("11.243.32.178,11.243.32.179,11.243.32.177", "2181", "/export/kerberos/1/hbase.keytab", "/etc/krb5.conf", "hbase/ocean@PUDAO.COM");

                if (HTableConst.INFO_TYPE_MOBILE.equals(info_type) && HTableConst.HASH_MD5.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.MOBILE_MD5_TABLE));
                    colfamily=HTableConst.COL_FAMILY_MOBILE;
                }else if(HTableConst.INFO_TYPE_MOBILE.equals(info_type) && HTableConst.HASH_SHA256.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.MOBILE_SHA256_TABLE));
                    colfamily=HTableConst.COL_FAMILY_MOBILE;
                }else if(HTableConst.INFO_TYPE_MOBILE.equals(info_type) && HTableConst.HASH_SM3.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.MOBILE_SM3_TABLE));
                    colfamily=HTableConst.COL_FAMILY_MOBILE;
                }else if(HTableConst.INFO_TYPE_CERT.equals(info_type) && HTableConst.HASH_MD5.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.CERT_MD5_TABLE));
                    colfamily=HTableConst.COL_FAMILY_CERT;
                }else if(HTableConst.INFO_TYPE_CERT.equals(info_type) && HTableConst.HASH_SHA256.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.CERT_SHA256_TABLE));
                    colfamily=HTableConst.COL_FAMILY_CERT;
                }else if(HTableConst.INFO_TYPE_CERT.equals(info_type) && HTableConst.HASH_SM3.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.CERT_SM3_TABLE));
                    colfamily=HTableConst.COL_FAMILY_CERT;
                }else {
                    throw new NoTableException("无对应表信息，请核对传递参数是否正确");
                }

                System.out.println("查询hbase表为："+table+",列簇为："+colfamily);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        try{
            System.out.println("开始取数-->str:"+str+"，colfamily："+colfamily);
            byte[] row = new byte[]{};
            row = str.getBytes();
            System.out.println("step1");
            Get get = new Get(row);
            System.out.println("step2:"+table.getName().getNameAsString());
            Result result = table.get(get);
            System.out.println("step3"+result.toString());
            String valueStr = new String(result.getValue(Bytes.toBytes(colfamily),Bytes.toBytes(HTableConst.CELL)),"utf-8");
            System.out.println("查询结果为："+valueStr);
            return valueStr;
        }catch (Exception e){
            System.out.println("捕获异常");
            e.printStackTrace();
        }/*finally {
            try {
                if (connection != null) connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }*/
        return null;
    }

}
